package org.promote.hotspot.client.pusher;

import io.netty.channel.Channel;
import lombok.extern.java.Log;
import org.promote.hotspot.client.ClientContext;
import org.promote.hotspot.common.model.HotKeyModel;
import org.promote.hotspot.common.model.HotSpotMsg;
import org.promote.hotspot.common.model.KeyCountModel;
import org.promote.hotspot.common.model.MessageType;
import org.promote.hotspot.client.core.server.ServerInfoHolder;
import org.promote.hotspot.common.tool.SystemClock;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * netty推送
 *
 * @author enping.jep
 * @date 2023/11/29 14:02
 **/
@Log
public class NettyHotPusher implements HotPusher {
    @Override
    public void sendHotKey(String appName, List<HotKeyModel> list) {
        Map<Channel, List<HotKeyModel>> map = new HashMap<>();
        for (HotKeyModel model : list) {
            model.setCreateTime(SystemClock.now());
            //根据hash算法获取key对应的服务器，分发到对应服务器相应的Channel连接
            Channel channel = ServerInfoHolder.chooseChannel(model.getKey());
            if (channel == null) {
                continue;
            }
            //computeIfAbsent() 方法对 hashMap 中指定 key 的值进行重新计算，如果不存在这个 key，则添加到 hashMap 中
            //如果 key 对应的 value 不存在，则使用获取 function 重新计算后的值，并保存为该 key 的 value，否则返回 value。
            List<HotKeyModel> newList = map.computeIfAbsent(channel, k -> {
                return new ArrayList<>();
            });
            newList.add(model);
        }

        for (Channel channel : map.keySet()) {
            List<HotKeyModel> batch = map.get(channel);
            HotSpotMsg hotSpotMsg = new HotSpotMsg(MessageType.REQUEST_NEW_KEY, ClientContext.APP_NAME);
            hotSpotMsg.setHotKeyModels(batch);
            try {
                log.info("send hot key : " + hotSpotMsg);
                channel.writeAndFlush(hotSpotMsg).sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
                InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
                log.warning("flush error :" + inetSocketAddress.getAddress().getHostAddress());
            }
        }
    }

    @Override
    public void sendHotCount(String appName, List<KeyCountModel> list) {
        long now = SystemClock.now();
        Map<Channel, List<KeyCountModel>> map = new HashMap<>();
        for (KeyCountModel model : list) {
            model.setCreateTime(now);
            Channel channel = ServerInfoHolder.chooseChannel(model.getRuleKey());
            if (channel == null) {
                continue;
            }
            List<KeyCountModel> newList = map.computeIfAbsent(channel, k -> new ArrayList<>());
            newList.add(model);
        }

        for (Channel channel : map.keySet()) {
            List<KeyCountModel> batch = map.get(channel);
            HotSpotMsg hotSpotMsg = new HotSpotMsg(MessageType.REQUEST_HIT_COUNT, ClientContext.APP_NAME);
            hotSpotMsg.setKeyCountModels(batch);
            try {
                log.info("send hot count : " + hotSpotMsg);
                channel.writeAndFlush(hotSpotMsg).sync();
            } catch (InterruptedException e) {
                InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
                log.warning("flush error :" + inetSocketAddress.getAddress().getHostAddress());
            }
        }
    }
}
